from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode, MapFunction, FlatMapFunction, \
    RuntimeContext

# 1、创建flink执行环境
env = StreamExecutionEnvironment.get_execution_environment()

# 修改并行度
env.set_parallelism(1)

# 2、读取数据,得到DataStream，相当于RDD  （有界流）
lines_ds = env.read_text_file("G:\LanZhiPeiXun\Flink\data\words.txt")

# 1、使用lambda表达式方式
words_ds = lines_ds.flat_map(lambda line: line.split(","))


# words_ds.print()

# 2、使用类和对象方式
class WordFlatMapFunction(FlatMapFunction):

    # open方法，每一个task执行一次,再task启动的时候执行，一般用于创建数据库连接
    def open(self, runtime_context: RuntimeContext):
        print("open")

    # close方法，每一个task执行一次。在task关闭执行，一般用于关闭数据库连接
    def close(self):
        print("close")

    # 每一行数据执行一次，返回多个结果
    def flat_map(self, line):
        for word in line.split(","):
            yield word


words_ds = lines_ds.flat_map(WordFlatMapFunction())

words_ds.print()

env.execute()
